package org.zhadoop.streaming.operator;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang.StringUtils;
import org.zhadoop.streaming.config.StreamingConfig;
import org.zhadoop.streaming.event.IEventType;
import org.zhadoop.streaming.event.TupleEvent;
import org.zhadoop.streaming.exception.StreamingException;
import org.zhadoop.streaming.util.StreamingUtils;

/**
 * 功能性算子的接口
 * @author zwd
 *
 */
public class FunctionOperator extends AbsOperator{

	
	/**
	 * 
	 */
	private static final long serialVersionUID = -6284728521607463298L;
	
	private IEventType outputSchema;
	
	private String outputStreamName;
	
	private List<String> inputStreams;
	
	private Map<String,IEventType> inputSchemas;
	
	public FunctionOperator() {
		inputStreams= new ArrayList<String>();
		inputSchemas = new HashMap<>();
	}

	@Override
	public List<String> getInputStream() {
		return null;
	}

	@Override
	public void setConfig(StreamingConfig conf) throws StreamingException {
		// TODO Auto-generated method stub
		super.setConfig(conf);
//		this.addInputStream((String)conf.get(StreamingConfig.STREAMING_INNER_INPUT_STREAM_NAME));
//		this.addInputSchema((String)conf.get(StreamingConfig.STREAMING_INNER_INPUT_STREAM_NAME),
//                StreamingUtils.deSerializeSchema((String) conf.get(StreamingConfig.STREAMING_INNER_INPUT_SCHEMA)));
//        this.outputSchema = StreamingUtils.deSerializeSchema((String)(conf.get(StreamingConfig.STREAMING_INNER_OUTPUT_SCHEMA)));
//        this.outputStreamName = (String)conf.get(StreamingConfig.STREAMING_INNER_OUTPUT_STREAM_NAME);
	}
	
	@Override
	public String getOutputStream() {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public Map<String, IEventType> getInputSchema() {
		// TODO Auto-generat0
		return null;
	}

	@Override
	public IEventType getOutputSchema() {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public Map<String, GroupInfo> getGroupInfo() {
		// TODO Auto-generated method stub
		return null;
	}

	@Override
	public void execute(String streamName, TupleEvent event) throws StreamingException {
		// TODO Auto-generated method stub
		
	}

	@Override
	public void destory() throws StreamingException {
		// TODO Auto-generated method stub
		
	}

	@Override
	public void initialize() throws StreamingException {
		// TODO Auto-generated method stub
		
	}

	@Override
	public void setInputStream(List<String> streamNames) throws StreamingException {
		// TODO Auto-generated method stub
		
	}

	@Override
	public void setOutputStream(String streamName) throws StreamingException {
		// TODO Auto-generated method stub
		
	}

	@Override
	public void setInputSchema(Map<String, IEventType> schemas) throws StreamingException {
		// TODO Auto-generated method stub
		
	}

	@Override
	public void setOutputSchema(IEventType schema) throws StreamingException {
		// TODO Auto-generated method stub
		
	}
	
	/**
	 * 添加输入流
	 * @param streamName
	 */
	public void addInputStream(String streamName){
		if(!StringUtils.isEmpty(streamName)){
			if(!inputStreams.contains(streamName)){
				inputStreams.add(streamName);
			}
		}
	}
	
}
